
package com.joysuch.wwyt.mh.storage.impl;

import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import com.google.common.collect.Lists;
import com.joysuch.wwyt.core.common.util.JsonUtil;
import com.joysuch.wwyt.core.entity.BaseConfig;
import com.joysuch.wwyt.core.enums.BaseConfigCodes;
import com.joysuch.wwyt.core.enums.Status;
import com.joysuch.wwyt.core.exception.ApiBusinessException;
import com.joysuch.wwyt.core.service.BaseConfigService;
import com.joysuch.wwyt.core.support.Context;
import com.joysuch.wwyt.mh.DateUtils;
import com.joysuch.wwyt.mh.GZIPUtil;
import com.joysuch.wwyt.mh.bean.MhMonitoringPointDtoo;
import com.joysuch.wwyt.mh.entity.MhAlarmLimitConfig;
import com.joysuch.wwyt.mh.entity.MhMonitoringPoint;
import com.joysuch.wwyt.mh.repository.MhMonitoringPointDao;
import com.joysuch.wwyt.mh.service.MhMonitoringRealTimeService;
import com.joysuch.wwyt.mh.storage.IMonitorPointDataHandler;
import com.joysuch.wwyt.mh.storage.IMonitorPointDataService;
import com.joysuch.wwyt.mh.storage.MonitorPointData;
import com.joysuch.wwyt.mh.task.WriteCsvTaskForList;
import com.joysuch.wwyt.risk.entity.RiskEvaluationControl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.InfluxDB;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ResourceUtils;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.OutputStream;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author qingzhou
 * 2019-11-20
 */
@Slf4j
@Service
@RestController
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class MonitorPointDataServiceImpl implements IMonitorPointDataService {

    private ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private final InfluxDB influxDB;

    @Resource
    private IMonitorPointDataHandler defaultIMonitorPointDataHandler;
    @Resource
    private BaseConfigService baseConfigService;
    @Resource
    private MhMonitoringPointDao monitoringPointDao;
    @Autowired
    private MhMonitoringRealTimeService mhMonitoringRealTimeService;

    @Value("${wwyt.monitor.point.data.backup}")
    private String monitorPointDataBackupPath;

    // flag, 0 本地; 1 异地
    @Override
    public List<MonitorPointData> historyVals(Long tenantId, String orgCode, String tag, String startTime,
                                              String endTime, Integer flag) {
        MhMonitoringPoint point = monitoringPointDao.findByTenentIdAndOrgCodeAndPointCode(tenantId, orgCode, tag);
        if (point != null && (point.getState() == null || point.getState() == 0)) {
            // 监测点已关闭
            return Collections.EMPTY_LIST;
        }
        String timeInterval = "1m";
        long interval = 60;
        DateTimeFormatter sdf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime ts = LocalDateTime.parse(startTime, sdf);
        LocalDateTime te = LocalDateTime.parse(endTime, sdf);
        Duration between = Duration.between(ts, te);
        long toMinutes = between.toMinutes();
        if (toMinutes > 60 * 1440) {
            timeInterval = "120m";
            interval = 120 * 60;
        } else if (toMinutes > 30 * 1440) {
            timeInterval = "60m";
            interval = 60 * 60;
        } else if (toMinutes > 1440) {
            timeInterval = "4m";
            interval = 4 * 60;
        } else if (toMinutes > 360) {
            timeInterval = "2m";
            interval = 2 * 60;
        }

        if (flag == 0) {
            String sql = String.format(
                    "select time,  value  from \"%s\" where time >= '%s' and time <= '%s' and tagId='%s'  tz('Asia/Shanghai')",
                    getMeasurementName(tenantId, orgCode), startTime, endTime, tag.toUpperCase().replace("\\", "\\\\"));
            QueryResult query = influxDB.query(new Query(sql));
            List<QueryResult.Result> results = query.getResults();
            return readResults(tenantId, orgCode, tag, results);
        } else if (flag == 1) {
            HttpHeaders headers = new HttpHeaders();
            MediaType mtype = MediaType.parseMediaType("text/plain; charset=utf-8");
            headers.setContentType(mtype);
            StringBuilder builder = new StringBuilder();
            builder.append("[");
            builder.append("{").append("\"tag\":").append("\"").append(tag.replace("\\", "\\\\")).append("\",");
            builder.append("\"starttime\":").append(ts.toEpochSecond(ZoneOffset.of("+0"))).append(",");
            builder.append("\"endtime\":").append(te.toEpochSecond(ZoneOffset.of("+0"))).append(",");
            builder.append("\"interval\":").append(interval).append("}");
            builder.append("]");
            log.info("body:" + builder.toString());
            HttpEntity<String> readEntity = new HttpEntity<>(builder.toString(), headers);
            ParameterizedTypeReference<String> readRespType = new ParameterizedTypeReference<String>() {
            };
            BaseConfig config = baseConfigService.findConfig(tenantId, orgCode,
                    BaseConfigCodes.MONITOR_POINT_HTTP_SERVER_URL);
            RestTemplate restTemplate = new RestTemplate();
            ResponseEntity<String> readResult = restTemplate.exchange(config.getValue() + "/READ", HttpMethod.PUT,
                    readEntity, readRespType);
            String body = readResult.getBody();
            List<MhMonitoringPointDtoo> mhMonitoringPointDtoos = JsonUtil.fromListJson(body,
                    MhMonitoringPointDtoo.class);
            log.info("read size {}", mhMonitoringPointDtoos.size());
            List<MonitorPointData> monitorPointData = new ArrayList<>();
            for (MhMonitoringPointDtoo mhMonitoringPointDtoo : mhMonitoringPointDtoos) {
                List<Map<String, Object>> values = mhMonitoringPointDtoo.getValues();
                monitorPointData.addAll(handleData(tenantId, orgCode, values, mhMonitoringPointDtoo.getTag()));
            }
            return monitorPointData;
        }
        return new ArrayList<>();
    }

    private List<MonitorPointData> handleData(Long tenantId, String orgCode, List<Map<String, Object>> data,
                                              String tag) {
        List<MonitorPointData> list = Lists.newArrayList();
        for (Map<String, Object> tagData : data) {
            MonitorPointData pd = new MonitorPointData();
            Double time = Double.valueOf(tagData.get("time").toString());
            BigDecimal bigDecimal = new BigDecimal(time);
            long l = bigDecimal.longValue();
            pd.setTime(l);
            pd.setTenentId(tenantId);
            pd.setOrgCode(orgCode);
            pd.setPointCode(tag);
            if (tagData.get("value") != null) {
                pd.setValue(Double.valueOf(tagData.get("value").toString()));
                list.add(pd);
            }
        }
        if (list.size() > 0) {
            Map<String, MhAlarmLimitConfig> configMap = new HashMap<>();
            Map<String,Map<Integer,String>> levelMaps = new HashMap<>();
            BaseConfig switchConfig = new BaseConfig();
            Map<String, RiskEvaluationControl> dyContMap = new HashMap<>();
            try {
                defaultIMonitorPointDataHandler.handle(list,configMap,levelMaps,switchConfig,dyContMap);
            } catch (Exception e) {
            }
        }
        return list;
    }

    private String getMeasurementName(Long tenantId, String orgCode) {
        return tenantId + "_" + orgCode;
    }

    @Override
    public Optional<MonitorPointData> realTimeVal(Long tenantId, String orgCode, String tag) {
        MhMonitoringPoint point = monitoringPointDao.findByTenentIdAndOrgCodeAndPointCode(tenantId, orgCode, tag);
        if (point != null && (point.getState() == null || point.getState() == 0)) {
            // 监测点已关闭
            return Optional.ofNullable(null);
        }

        QueryResult query = influxDB.query(new Query(String.format(
                "select time, last(value) as value from \"%s\" where time > now() - 5m and tagId='%s' tz('Asia/Shanghai')",
                getMeasurementName(tenantId, orgCode), tag.toUpperCase().replace("\\", "\\\\"))));
        List<QueryResult.Result> results = query.getResults();
        List<MonitorPointData> rs = readResults(tenantId, orgCode, tag, results);
        for (MonitorPointData data : rs) {
            String exValue = mhMonitoringRealTimeService.getVauleByPointsUnits(point, data.getValue().toString());
            data.setExValue(exValue);
        }
        return Optional.ofNullable(rs.size() > 0 ? rs.get(0) : null);
    }


    @Override
    public void monitorPointDataBackup(HttpServletResponse response, Date startTime, Date endTime) throws ApiBusinessException {
        Long tenantId = Context.getCurrentTenantId();
        String orgCode = Context.getCompanyCode();
        Date startDate = DateUtil.beginOfDay(startTime);
        Date endDate = DateUtil.endOfDay(endTime);
        List<MhMonitoringPoint> pointList = monitoringPointDao.findByTenentIdAndOrgCode(tenantId, orgCode);
        String zipFileName = "MonitorPoint_" + DateUtil.format(new Date(), DateUtils.STYLE_yyyyMMddHHmmss);
        final String basePathname = monitorPointDataBackupPath + zipFileName;
        Phaser phaser = new Phaser(1);
        for (MhMonitoringPoint each : pointList) {
            QueryResult query = influxDB.query(new Query(String.format(
                    "select * from \"%s\" where tagId='%s' and time >= '%s' and time < '%s' tz('Asia/Shanghai')",
                    getMeasurementName(tenantId, orgCode), each.getPointCode().toUpperCase().replace("\\", "\\\\"), startDate, endDate)));
            List<QueryResult.Result> results = query.getResults();
            if (!CollectionUtils.isEmpty(results)) {
                Optional<QueryResult.Result> optionalResult = results.stream().filter(e -> !CollectionUtils.isEmpty(e.getSeries())).findFirst();
                if (optionalResult.isPresent()) {
                    phaser.register();
                    String filePath = basePathname + File.separator + each.getPointCode() + File.separator + DateUtil.format(startDate, DateUtils.STYLE_yyyyMMdd) + "_" + DateUtil.format(endDate, DateUtils.STYLE_yyyyMMdd) + File.separator + each.getPointCode();
                    executor.submit(new WriteCsvTaskForList(filePath, results, phaser));
                }
            }
        }
        phaser.arriveAndDeregister();
        String zipFile = null;
        try {
            while (phaser.getUnarrivedParties() != 0) {
                Thread.sleep(500);
            }
            zipFile = GZIPUtil.zipFile(new File(basePathname), "zip");
            //response.setContentType("APPLICATION/OCTET-STREAM");
            response.setHeader("Content-Disposition", "attachment; filename=" + zipFileName + ".zip");
            OutputStream out = response.getOutputStream();
            File ftp = ResourceUtils.getFile(zipFile);
            InputStream in = new FileInputStream(ftp);
            byte[] b = new byte[1024];
            int len;
            while ((len = in.read(b)) != -1) {
                out.write(b, 0, len);
            }
            in.close();
            out.close();
        } catch (Exception e) {
            log.error("Compress file error : ", e);
            throw new ApiBusinessException(Status.EXPORT_MONITOR_DATA_ERROR);
        } finally {
            FileUtil.del(basePathname);
            /*if (zipFile != null) {
                FileUtil.del(zipFile);
            }*/
        }
    }

    @Override
    public boolean getMonitorPointBackupFileSize(Date startTime, Date endTime) throws ApiBusinessException {
        boolean fileExist = false;
        Long tenantId = Context.getCurrentTenantId();
        String orgCode = Context.getCompanyCode();
        Date startDate = DateUtil.beginOfDay(startTime);
        Date endDate = DateUtil.endOfDay(endTime);
        long betweenDay = DateUtil.between(endDate, startDate, DateUnit.DAY);
        if (betweenDay > 90) {
            throw new ApiBusinessException(Status.EXPORT_MONITOR_DATA_LIMIT_ERROR);
        }
        List<MhMonitoringPoint> pointList = monitoringPointDao.findByTenentIdAndOrgCode(tenantId, orgCode);
        for (MhMonitoringPoint each : pointList) {
            QueryResult query = influxDB.query(new Query(String.format(
                    "select * from \"%s\" where tagId='%s' and time >= '%s' and time < '%s' tz('Asia/Shanghai')",
                    getMeasurementName(tenantId, orgCode), each.getPointCode().toUpperCase().replace("\\", "\\\\"), startDate, endDate)));
            List<QueryResult.Result> results = query.getResults();
            if (!CollectionUtils.isEmpty(results)) {
                Optional<QueryResult.Result> optionalResult = results.stream().filter(e -> !CollectionUtils.isEmpty(e.getSeries())).findFirst();
                if (optionalResult.isPresent()) {
                    fileExist = true;
                    break;
                }
            }
        }
        return fileExist;
    }

    /**
     * 批量获取监测点最近一次数据
     *
     * @param tenantId
     * @param orgCode
     * @param tagCods
     * @return
     */
    @Override
    public List<MonitorPointData> getRealTimeVals(Long tenantId, String orgCode, List<String> tagCods) {
        if (CollectionUtils.isEmpty(tagCods)) {
            return null;
        }
        List<QueryResult.Result> results = new ArrayList<>();
        //分批查询
        List<List<String>> tagCodsList = Lists.partition(tagCods,1000);
        for (List<String> tagIds : tagCodsList){
            StringBuilder builder = new StringBuilder("select time, last(value) as value from \"");
            builder.append(getMeasurementName(tenantId, orgCode)).append("\" where time > now() - 5m and ");
            for (String tag : tagIds) {
                builder.append(" tagId ='" + tag.replace("\\", "\\\\") + "' or");
            }
            builder.append(" 1=1 group by tagId tz('Asia/Shanghai')");
            QueryResult query = influxDB.query(new Query(builder.toString()));
            results.addAll(query.getResults());
        }
        List<MonitorPointData> rs = readResults(tenantId, orgCode, tagCods, results);
        return rs;
    }

    private List<MonitorPointData> readResults(Long tenantId, String orgCode, List<String> tagCods,
                                               List<QueryResult.Result> results) {
        List<MonitorPointData> ls = Lists.newArrayList();

        if (results != null) {
            for (QueryResult.Result r : results) {
                List<QueryResult.Series> series = r.getSeries();
                if (series == null) {
                    continue;
                }
                for (String tag : tagCods) {
                    for (QueryResult.Series ser : series) {
                        //字段 time ,value
                        List<String> columns = ser.getColumns();
                        //字段对应的值
                        List<List<Object>> records = ser.getValues();
                        //tag
                        Map<String, String> tags = ser.getTags();
                        if (records.size() == 0) {
                            continue;
                        }
                        if (tags != null && tags.get("tagId").equalsIgnoreCase(tag)) {
                            this.crateDataBean(ls, tag, tenantId, orgCode, columns, records);
                            continue;
                        }
                    }

                }
            }
        }

        return ls;
    }

    //data->bean
    private void crateDataBean(List<MonitorPointData> ls, String tag, Long tenantId, String
            orgCode, List<String> columns, List<List<Object>> records) {
        int timeIdx = -1;
        int valIdx = -1;
        for (int i = 0; i < columns.size(); i++) {
            if ("value".equals(columns.get(i))) {
                valIdx = i;
            } else if ("time".equals(columns.get(i))) {
                timeIdx = i;
            }
        }
        if (timeIdx != -1 && valIdx != -1) {
            for (List<Object> record : records) {
                MonitorPointData p = new MonitorPointData();
                String timeStr = (String) record.get(timeIdx);
                timeStr = timeStr.substring(0, timeStr.lastIndexOf("+") - 1);
                Double valStr = (Double) record.get(valIdx);
                p.setTenentId(tenantId);
                p.setOrgCode(orgCode);
                p.setPointCode(tag);
                p.setValue(valStr);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
                try {
                    Date parse = sdf.parse(timeStr);
                    p.setTime(parse.getTime() / 1000);
                } catch (ParseException e) {
                    p.setTime(-1L);
                }
                ls.add(p);
            }
        }
    }

//    //查询不到数据的point也返回 值为null
//    private void createDefaultBean(List<MonitorPointData> ls, String tag, Long tenantId, String orgCode) {
//        MonitorPointData data = new MonitorPointData();
//        data.setOrgCode(orgCode);
//        data.setTenentId(tenantId);
//        data.setPointCode(tag);
//        ls.add(data);
//    }

    private List<MonitorPointData> readResults(Long tenantId, String orgCode, String tag,
                                               List<QueryResult.Result> results) {
        List<MonitorPointData> ls = Lists.newArrayList();
        if (results != null) {
            for (QueryResult.Result r : results) {
                List<QueryResult.Series> series = r.getSeries();
                if (series == null) {
                    continue;
                }
                for (QueryResult.Series ser : series) {
                    //字段 time ,value
                    List<String> columns = ser.getColumns();
                    //字段对应的值
                    List<List<Object>> records = ser.getValues();
                    //tag
                    Map<String, String> tags = ser.getTags();
                    if (records.size() == 0) {
                        continue;
                    }
                    int timeIdx = -1;
                    int valIdx = -1;
                    for (int i = 0; i < columns.size(); i++) {
                        if ("value".equals(columns.get(i))) {
                            valIdx = i;
                        } else if ("time".equals(columns.get(i))) {
                            timeIdx = i;
                        }
                    }
                    if (timeIdx != -1 && valIdx != -1) {
                        for (List<Object> record : records) {
                            MonitorPointData p = new MonitorPointData();
                            String timeStr = (String) record.get(timeIdx);
                            if (timeStr.contains("+")) {
                                timeStr = timeStr.substring(0, timeStr.lastIndexOf("+") - 1);
                            }
                            Double valStr = (Double) record.get(valIdx);
                            p.setTenentId(tenantId);
                            p.setOrgCode(orgCode);
                            if (tags != null && tags.get("tagId") != null) {
                                p.setPointCode(tags.get("tagId"));
                            } else {
                                p.setPointCode(tag);
                            }
                            p.setValue(valStr);
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
                            try {
                                Date parse = sdf.parse(timeStr);
                                p.setTime(parse.getTime() / 1000);
                            } catch (ParseException e) {
                                p.setTime(-1L);
                            }
                            ls.add(p);
                        }
                    }
                }
            }
        }
        return ls;
    }


}
